package com.luxf.base.domainevent.rocketmq;

import com.luxf.base.domainevent.DomainEventNoticeHandler;
import com.luxf.base.domainevent.DomainEventSubscriber;
import com.luxf.base.domainevent.codec.MessageDecoder;
import com.luxf.base.domainevent.model.DomainEventNotice;
import com.luxf.base.domainevent.utils.EventNoticeUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQListener;

import javax.annotation.Resource;

@Slf4j
public class RocketDomainEventNoticeListener<T> implements RocketMQListener<T>, DomainEventSubscriber {

    @Getter
    private RocketMessageListenerProperties rocketMessageListenerProperties;

    @Resource
    private MessageDecoder messageDecoder;

    @Resource
    private DomainEventNoticeHandler domainEventNoticeHandler;

    @Override
    public void onMessage(T message) {
        DomainEventNotice domainEventNotice = messageDecoder.decode(message.toString(), DomainEventNotice.class);
        log.info("收到领域通知id:{}", EventNoticeUtils.getMessageId(domainEventNotice));
        consume(domainEventNotice);
    }

    private void consume(DomainEventNotice eventNotice) {
        domainEventNoticeHandler.handle(eventNotice, (notice, isCommit) -> {
            if (!isCommit) {
                throw new RuntimeException("消息消费失败");
            }
        });
    }


    public static RocketDomainEventNoticeListener<?> create(RocketMessageListenerProperties rocketMessageListenerProperties) {
        RocketDomainEventNoticeListener<?> noticeListener = new RocketDomainEventNoticeListener<>();
        noticeListener.rocketMessageListenerProperties = rocketMessageListenerProperties;
        return noticeListener;
    }


}
